package com.fwh.web.rabbitmq;

import com.rabbitmq.client.Channel;
import com.sun.org.apache.xpath.internal.operations.String;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * mq消费者监听
 *
 * @author fwh
 * @email fuwenhao594@163.com
 * @date 2019/11/15 上午11:26
 */
@Component
@Slf4j
public class RabbitMqConsumer implements ChannelAwareMessageListener {

//    @Autowired
//    private ConnectionFactory connectionFactory;
//
//    @Bean
//    public SimpleMessageListenerContainer execMessageContainerOne() {
//        //设置监听者“容器”
//        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
//        log.info("通知mq消费方启动监听队列名{}", "fwh");
//        //设置队列名
//        container.setQueueNames("fwh");
//        //设置监听者数量，即消费线程数
//        container.setConcurrentConsumers(1);
//        container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
//            if (null != message && null != message.getBody()) {
//                try {
//                    //实际业务处理方法
//                    System.out.println(message);
////                    notifyReceiver.notifyProcess(message);
//                } catch (Exception e) {
//                    log.error("通知模块启动监听异常");
//                }
//            }
//        });
//        return container;
//    }





    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        /*//消费端正常处理完成一条mq时，ack mq服务器可以移除此条mq的方法
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        //消费端处理完成一条mq时发生异常，ack 会将此条mq重新放到mq服务器队列queue中
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        //就是消费端主动拒绝mq服务器发送mq过来。
        channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);*/

        System.out.println("接收的消息为：" + message.getBody());
        //业务方法处理是否成功
        Boolean mqFlag = true;
        if (mqFlag) {
            basicACK(message, channel);//处理正常--ack
        } else {
            basicNACK(message, channel);//处理异常--nack
        }
    }

    //正常消费掉后通知mq服务器移除此条mq
    private void basicACK(Message message, Channel channel) {
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            log.error("通知服务器移除mq时异常，异常信息：" + e);
        }
    }

    //处理异常，mq重回队列
    private void basicNACK(Message message, Channel channel) {
        try {
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        } catch (IOException e) {
            log.error("mq重新进入服务器时出现异常，异常信息：" + e);
        }
    }
}
